/**
 * 
 */
package com.shelpe.services.sinbad.service.mq.impl;

import java.util.concurrent.CountDownLatch;

import javax.annotation.Resource;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;

import com.shelpe.services.sinbad.constvalue.ConstValue;
import com.shelpe.services.sinbad.constvalue.ConstValue.HOME_REPORT_SYNC_ENUM;
import com.shelpe.services.sinbad.controller.dto.redis.RedisDto;
import com.shelpe.services.sinbad.service.base.BaseService;
import com.shelpe.services.sinbad.service.mq.ShelpeMQConsumerService;
import com.shelpe.services.sinbad.service.mq.ShelpeMQProcessService;

/**
 * kafka 接收消息的实现类
 * @author louxinhua
 *
 */
public class ShelpeMQConsumerServiceImpl extends BaseService implements ShelpeMQConsumerService {

	private CountDownLatch latch = new CountDownLatch(1);

	/**
	 * 处理登录的数据同步
	 */
	@Resource(name="kafkaMPProcessForLogin")
	private ShelpeMQProcessService loginProcess;
	
	/**
	 * 处理自动优化的数据同步
	 */
	@Resource(name="kafkaMPProcessForOptimize")
	private ShelpeMQProcessService optimizeProcess;
	
	public CountDownLatch getLatch() {
		return latch;
	}
	
	@KafkaListener(topics = "${kafka.sinbad.homesynctopic}")
	public void receive(ConsumerRecord<?, ?> consumerRecord) {
		
		this.logger.debug("received payload='{}'", consumerRecord.toString());
		
		// TODO:需要解析consumerRecord中的内容
		RedisDto dto = this.unserialize((String)consumerRecord.value());
		HOME_REPORT_SYNC_ENUM type = null;
		if("manual".equals(dto.getType())){
			type = HOME_REPORT_SYNC_ENUM.login;
		}else {
			type = HOME_REPORT_SYNC_ENUM.auto_optimize;
		}
		
		//or TODO:根据consumerRecord的内容生成map
//		Map<String, String> params = null;
		
		// TODO:进行处理
		ShelpeMQProcessService processService = genProcess(type);
		
		processService.process(dto);
		
		
		latch.countDown();
	}
	
	
	
	/**
	 * 根据同步类型返回service bean
	 * @param type
	 */
	private ShelpeMQProcessService genProcess(ConstValue.HOME_REPORT_SYNC_ENUM type) {
		
		if ( HOME_REPORT_SYNC_ENUM.login.ordinal() == type.ordinal() ) {
			// 登录
			return loginProcess;
			
		}
		else {
			// 自动优化
			return optimizeProcess;
		}
		
	}
	
	
	
}
